package com.pool;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.*;

public class MyThreadPoolManager {

    public static void main(String[] args) {

        final boolean[] complete = {false};
        ExecutorService service = new ThreadPoolExecutor(
                2,
                8,
                1000,
                TimeUnit.SECONDS,
                new ArrayBlockingQueue<Runnable>(4),
                new RejectedExecutionHandler() {
                    @Override
                    public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
                        try {
                            executor.getQueue().put(r);
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }
                    }
                });

        LinkedBlockingQueue<String> queue = new LinkedBlockingQueue<>(100);
        new Thread(new Runnable() {
            @Override
            public void run() {
                for (int i = 0; i < 10; i++) {
                    try {
                        queue.put("data:" + System.currentTimeMillis());
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
                complete[0] = true;
            }
        }).start();

        List<Future> taskRes = new ArrayList<>();
        for (int i = 0; i < 4; i++) {
            Future<?> res = service.submit(new Runnable() {
                @Override
                public void run() {
                    while (true) {
                        try {
                            System.out.println(Thread.currentThread() + "=>" + queue.take());
                            if(complete[0]&&queue.size()==0){
                                System.exit(0);
                            }
                        } catch (InterruptedException e) {
                            System.out.println(Thread.currentThread() + "=>InterruptedException");
                            break;
                        }
                    }
                    System.out.println(Thread.currentThread() + "--------------exit-----------");
                }
            });

            taskRes.add(res);
        }



        new Thread(new Runnable() {
            @Override
            public void run() {
                while (true){
                    for (int i = 0; i < taskRes.size(); i++) {
                        try {
                            Thread.sleep(1000);
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }
                        System.out.println(taskRes.get(i).isDone());
                    }
                }
            }
        }).start();


        if(complete[0]){
            System.out.println("shutdown.....");
            service.shutdown();
        }

        try {
            while (service.awaitTermination(1000 * 1, TimeUnit.MILLISECONDS)) {
                System.out.println("waitting for exit......");
            }
        } catch (Exception e) {
            e.printStackTrace();
            System.out.println("interupted......");
        }

    }
}
